// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#pragma once

#include <algorithm>
#include <cstdint>
#include <memory>
#include <type_traits>

#include <nebula/array/array_base.h>
#include <nebula/array/array_binary.h>
#include <nebula/array/builder_adaptive.h>   // IWYU pragma: export
#include <nebula/array/builder_base.h>       // IWYU pragma: export
#include <nebula/array/builder_primitive.h>  // IWYU pragma: export
#include <nebula/array/data.h>
#include <nebula/array/util.h>
#include <nebula/types/scalar.h>
#include <turbo/utility/status.h>
#include <nebula/types/type.h>
#include <nebula/types/type_traits.h>
#include <nebula/bits/bit_block_counter.h>
#include <turbo/base/checked_cast.h>
#include <nebula/numeric/decimal.h>
#include <turbo/base/macros.h>

namespace nebula::internal {

    template<typename T, typename Enable = void>
    struct DictionaryValue {
        using type = typename T::c_type;
        using PhysicalType = T;
    };

    template<typename T>
    struct DictionaryValue<T, enable_if_base_binary<T>> {
        using type = std::string_view;
        using PhysicalType =
                typename std::conditional<std::is_same<typename T::offset_type, int32_t>::value,
                        BinaryType, LargeBinaryType>::type;
    };

    template<typename T>
    struct DictionaryValue<T, enable_if_binary_view_like<T>> {
        using type = std::string_view;
        using PhysicalType = BinaryViewType;
    };

    template<typename T>
    struct DictionaryValue<T, enable_if_fixed_size_binary<T>> {
        using type = std::string_view;
        using PhysicalType = BinaryType;
    };

    class TURBO_EXPORT DictionaryMemoTable {
    public:
        DictionaryMemoTable(MemoryPool *pool, const std::shared_ptr<DataType> &type);

        DictionaryMemoTable(MemoryPool *pool, const std::shared_ptr<Array> &dictionary);

        ~DictionaryMemoTable();

        turbo::Status GetArrayData(int64_t start_offset, std::shared_ptr<ArrayData> *out);

        /// \brief insert new memo values
        turbo::Status InsertValues(const Array &values);

        int32_t size() const;

        template<typename T>
        turbo::Status GetOrInsert(typename DictionaryValue<T>::type value, int32_t *out) {
            // We want to keep the DictionaryMemoTable implementation private, also we can't
            // use extern template classes because of compiler issues (MinGW?).  Instead,
            // we expose explicit function overrides for each supported physical type.
            const typename DictionaryValue<T>::PhysicalType *physical_type = nullptr;
            return GetOrInsert(physical_type, value, out);
        }

    private:
        turbo::Status GetOrInsert(const BooleanType *, bool value, int32_t *out);

        turbo::Status GetOrInsert(const Int8Type *, int8_t value, int32_t *out);

        turbo::Status GetOrInsert(const Int16Type *, int16_t value, int32_t *out);

        turbo::Status GetOrInsert(const Int32Type *, int32_t value, int32_t *out);

        turbo::Status GetOrInsert(const Int64Type *, int64_t value, int32_t *out);

        turbo::Status GetOrInsert(const UInt8Type *, uint8_t value, int32_t *out);

        turbo::Status GetOrInsert(const UInt16Type *, uint16_t value, int32_t *out);

        turbo::Status GetOrInsert(const UInt32Type *, uint32_t value, int32_t *out);

        turbo::Status GetOrInsert(const UInt64Type *, uint64_t value, int32_t *out);

        turbo::Status GetOrInsert(const DurationType *, int64_t value, int32_t *out);

        turbo::Status GetOrInsert(const TimestampType *, int64_t value, int32_t *out);

        turbo::Status GetOrInsert(const Date32Type *, int32_t value, int32_t *out);

        turbo::Status GetOrInsert(const Date64Type *, int64_t value, int32_t *out);

        turbo::Status GetOrInsert(const Time32Type *, int32_t value, int32_t *out);

        turbo::Status GetOrInsert(const Time64Type *, int64_t value, int32_t *out);

        turbo::Status GetOrInsert(const MonthDayNanoIntervalType *,
                                  MonthDayNanoIntervalType::MonthDayNanos value, int32_t *out);

        turbo::Status GetOrInsert(const DayTimeIntervalType *,
                                  DayTimeIntervalType::DayMilliseconds value, int32_t *out);

        turbo::Status GetOrInsert(const MonthIntervalType *, int32_t value, int32_t *out);

        turbo::Status GetOrInsert(const Fp32Type *, float value, int32_t *out);

        turbo::Status GetOrInsert(const Fp64Type *, double value, int32_t *out);

        turbo::Status GetOrInsert(const BinaryType *, std::string_view value, int32_t *out);

        turbo::Status GetOrInsert(const LargeBinaryType *, std::string_view value, int32_t *out);

        turbo::Status GetOrInsert(const BinaryViewType *, std::string_view value, int32_t *out);

        class DictionaryMemoTableImpl;

        std::unique_ptr<DictionaryMemoTableImpl> impl_;
    };


    /// \brief Array builder for created encoded DictionaryArray from
    /// dense array
    ///
    /// Unlike other builders, dictionary builder does not completely
    /// reset the state on finish calls.
    template<typename BuilderType, typename T>
    class DictionaryBuilderBase : public ArrayBuilder {
    public:
        using TypeClass = DictionaryType;
        using Value = typename DictionaryValue<T>::type;

        // WARNING: the type given below is the value type, not the DictionaryType.
        // The DictionaryType is instantiated on the finish() call.
        template<typename B = BuilderType, typename T1 = T>
        DictionaryBuilderBase(uint8_t start_int_size,
                              enable_if_t<std::is_base_of<AdaptiveIntBuilderBase, B>::value &&
                                          !is_fixed_size_binary_type<T1>::value,
                                      const std::shared_ptr<DataType> &>
                              value_type,
                              MemoryPool *pool = default_memory_pool(),
                              int64_t alignment = kDefaultBufferAlignment)
                : ArrayBuilder(pool, alignment),
                  memo_table_(new internal::DictionaryMemoTable(pool, value_type)),
                  delta_offset_(0),
                  byte_width_(-1),
                  indices_builder_(start_int_size, pool, alignment),
                  value_type_(value_type) {}

        template<typename T1 = T>
        explicit DictionaryBuilderBase(
                enable_if_t<!is_fixed_size_binary_type<T1>::value, const std::shared_ptr<DataType> &>
                value_type,
                MemoryPool *pool = default_memory_pool(),
                int64_t alignment = kDefaultBufferAlignment)
                : ArrayBuilder(pool, alignment),
                  memo_table_(new internal::DictionaryMemoTable(pool, value_type)),
                  delta_offset_(0),
                  byte_width_(-1),
                  indices_builder_(pool, alignment),
                  value_type_(value_type) {}

        template<typename T1 = T>
        explicit DictionaryBuilderBase(
                const std::shared_ptr<DataType> &index_type,
                enable_if_t<!is_fixed_size_binary_type<T1>::value, const std::shared_ptr<DataType> &>
                value_type,
                MemoryPool *pool = default_memory_pool(),
                int64_t alignment = kDefaultBufferAlignment)
                : ArrayBuilder(pool, alignment),
                  memo_table_(new internal::DictionaryMemoTable(pool, value_type)),
                  delta_offset_(0),
                  byte_width_(-1),
                  indices_builder_(index_type, pool, alignment),
                  value_type_(value_type) {}

        template<typename B = BuilderType, typename T1 = T>
        DictionaryBuilderBase(uint8_t start_int_size,
                              enable_if_t<std::is_base_of<AdaptiveIntBuilderBase, B>::value &&
                                          is_fixed_size_binary_type<T1>::value,
                                      const std::shared_ptr<DataType> &>
                              value_type,
                              MemoryPool *pool = default_memory_pool(),
                              int64_t alignment = kDefaultBufferAlignment)
                : ArrayBuilder(pool, alignment),
                  memo_table_(new internal::DictionaryMemoTable(pool, value_type)),
                  delta_offset_(0),
                  byte_width_(static_cast<const T1 &>(*value_type).byte_width()),
                  indices_builder_(start_int_size, pool, alignment),
                  value_type_(value_type) {}

        template<typename T1 = T>
        explicit DictionaryBuilderBase(
                enable_if_fixed_size_binary<T1, const std::shared_ptr<DataType> &> value_type,
                MemoryPool *pool = default_memory_pool(),
                int64_t alignment = kDefaultBufferAlignment)
                : ArrayBuilder(pool, alignment),
                  memo_table_(new internal::DictionaryMemoTable(pool, value_type)),
                  delta_offset_(0),
                  byte_width_(static_cast<const T1 &>(*value_type).byte_width()),
                  indices_builder_(pool, alignment),
                  value_type_(value_type) {}

        template<typename T1 = T>
        explicit DictionaryBuilderBase(
                const std::shared_ptr<DataType> &index_type,
                enable_if_fixed_size_binary<T1, const std::shared_ptr<DataType> &> value_type,
                MemoryPool *pool = default_memory_pool(),
                int64_t alignment = kDefaultBufferAlignment)
                : ArrayBuilder(pool, alignment),
                  memo_table_(new internal::DictionaryMemoTable(pool, value_type)),
                  delta_offset_(0),
                  byte_width_(static_cast<const T1 &>(*value_type).byte_width()),
                  indices_builder_(index_type, pool, alignment),
                  value_type_(value_type) {}

        template<typename T1 = T>
        explicit DictionaryBuilderBase(
                enable_if_parameter_free<T1, MemoryPool *> pool = default_memory_pool())
                : DictionaryBuilderBase<BuilderType, T1>(TypeTraits<T1>::type_singleton(), pool) {}

        // This constructor doesn't check for errors. Use InsertMemoValues instead.
        explicit DictionaryBuilderBase(const std::shared_ptr<Array> &dictionary,
                                       MemoryPool *pool = default_memory_pool(),
                                       int64_t alignment = kDefaultBufferAlignment)
                : ArrayBuilder(pool, alignment),
                  memo_table_(new internal::DictionaryMemoTable(pool, dictionary)),
                  delta_offset_(0),
                  byte_width_(-1),
                  indices_builder_(pool, alignment),
                  value_type_(dictionary->type()) {}

        ~DictionaryBuilderBase() override = default;

        /// \brief The current number of entries in the dictionary
        int64_t dictionary_length() const { return memo_table_->size(); }

        /// \brief The value byte width (for FixedSizeBinaryType)
        template<typename T1 = T>
        enable_if_fixed_size_binary<T1, int32_t> byte_width() const {
            return byte_width_;
        }

        /// \brief append a scalar value
        turbo::Status append(Value value) {
            TURBO_RETURN_NOT_OK(Reserve(1));

            int32_t memo_index;
            TURBO_RETURN_NOT_OK(memo_table_->GetOrInsert<T>(value, &memo_index));
            TURBO_RETURN_NOT_OK(indices_builder_.append(memo_index));
            length_ += 1;

            return turbo::OkStatus();
        }

        /// \brief append a fixed-width string (only for FixedSizeBinaryType)
        template<typename T1 = T>
        enable_if_fixed_size_binary<T1, turbo::Status> append(const uint8_t *value) {
            return append(std::string_view(reinterpret_cast<const char *>(value), byte_width_));
        }

        /// \brief append a fixed-width string (only for FixedSizeBinaryType)
        template<typename T1 = T>
        enable_if_fixed_size_binary<T1, turbo::Status> append(const char *value) {
            return append(std::string_view(value, byte_width_));
        }

        /// \brief append a string (only for binary types)
        template<typename T1 = T>
        enable_if_binary_like<T1, turbo::Status> append(const uint8_t *value, int32_t length) {
            return append(reinterpret_cast<const char *>(value), length);
        }

        /// \brief append a string (only for binary types)
        template<typename T1 = T>
        enable_if_binary_like<T1, turbo::Status> append(const char *value, int32_t length) {
            return append(std::string_view(value, length));
        }

        /// \brief append a string (only for string types)
        template<typename T1 = T>
        enable_if_string_like<T1, turbo::Status> append(const char *value, int32_t length) {
            return append(std::string_view(value, length));
        }

        /// \brief append a decimal (only for Decimal128Type)
        template<typename T1 = T>
        enable_if_decimal128<T1, turbo::Status> append(const Decimal128 &value) {
            uint8_t data[16];
            value.ToBytes(data);
            return append(data, 16);
        }

        /// \brief append a decimal (only for Decimal128Type)
        template<typename T1 = T>
        enable_if_decimal256<T1, turbo::Status> append(const Decimal256 &value) {
            uint8_t data[32];
            value.ToBytes(data);
            return append(data, 32);
        }

        /// \brief append a scalar null value
        turbo::Status append_null() final {
            length_ += 1;
            null_count_ += 1;

            return indices_builder_.append_null();
        }

        turbo::Status append_nulls(int64_t length) final {
            length_ += length;
            null_count_ += length;

            return indices_builder_.append_nulls(length);
        }

        turbo::Status append_empty_value() final {
            length_ += 1;

            return indices_builder_.append_empty_value();
        }

        turbo::Status append_empty_values(int64_t length) final {
            length_ += length;

            return indices_builder_.append_empty_values(length);
        }

        turbo::Status append_scalar(const Scalar &scalar, int64_t n_repeats) override {
            if (!scalar.is_valid) return append_nulls(n_repeats);

            const auto &dict_ty = turbo::checked_cast<const DictionaryType &>(*scalar.type);
            const DictionaryScalar &dict_scalar =
                    turbo::checked_cast<const DictionaryScalar &>(scalar);
            const auto &dict = turbo::checked_cast<const typename TypeTraits<T>::ArrayType &>(
                    *dict_scalar.value.dictionary);
            TURBO_RETURN_NOT_OK(Reserve(n_repeats));
            switch (dict_ty.index_type()->id()) {
                case Type::UINT8:
                    return AppendScalarImpl<UInt8Type>(dict, *dict_scalar.value.index, n_repeats);
                case Type::INT8:
                    return AppendScalarImpl<Int8Type>(dict, *dict_scalar.value.index, n_repeats);
                case Type::UINT16:
                    return AppendScalarImpl<UInt16Type>(dict, *dict_scalar.value.index, n_repeats);
                case Type::INT16:
                    return AppendScalarImpl<Int16Type>(dict, *dict_scalar.value.index, n_repeats);
                case Type::UINT32:
                    return AppendScalarImpl<UInt32Type>(dict, *dict_scalar.value.index, n_repeats);
                case Type::INT32:
                    return AppendScalarImpl<Int32Type>(dict, *dict_scalar.value.index, n_repeats);
                case Type::UINT64:
                    return AppendScalarImpl<UInt64Type>(dict, *dict_scalar.value.index, n_repeats);
                case Type::INT64:
                    return AppendScalarImpl<Int64Type>(dict, *dict_scalar.value.index, n_repeats);
                default:
                    return turbo::failed_precondition_error("Invalid index type: ", dict_ty);
            }
            return turbo::OkStatus();
        }

        turbo::Status append_scalars(const ScalarVector &scalars) override {
            for (const auto &scalar: scalars) {
                TURBO_RETURN_NOT_OK(append_scalar(*scalar, /*n_repeats=*/1));
            }
            return turbo::OkStatus();
        }

        turbo::Status append_array_slice(const ArraySpan &array, int64_t offset, int64_t length) final {
            // Visit the indices and insert the unpacked values.
            const auto &dict_ty = turbo::checked_cast<const DictionaryType &>(*array.type);
            // See if possible to avoid using to_array_data here
            const typename TypeTraits<T>::ArrayType dict(array.dictionary().to_array_data());
            TURBO_RETURN_NOT_OK(Reserve(length));
            switch (dict_ty.index_type()->id()) {
                case Type::UINT8:
                    return AppendArraySliceImpl<uint8_t>(dict, array, offset, length);
                case Type::INT8:
                    return AppendArraySliceImpl<int8_t>(dict, array, offset, length);
                case Type::UINT16:
                    return AppendArraySliceImpl<uint16_t>(dict, array, offset, length);
                case Type::INT16:
                    return AppendArraySliceImpl<int16_t>(dict, array, offset, length);
                case Type::UINT32:
                    return AppendArraySliceImpl<uint32_t>(dict, array, offset, length);
                case Type::INT32:
                    return AppendArraySliceImpl<int32_t>(dict, array, offset, length);
                case Type::UINT64:
                    return AppendArraySliceImpl<uint64_t>(dict, array, offset, length);
                case Type::INT64:
                    return AppendArraySliceImpl<int64_t>(dict, array, offset, length);
                default:
                    return turbo::failed_precondition_error("Invalid index type: ", dict_ty);
            }
            return turbo::OkStatus();
        }

        /// \brief insert values into the dictionary's memo, but do not append any
        /// indices. Can be used to initialize a new builder with known dictionary
        /// values
        /// \param[in] values dictionary values to add to memo. Type must match
        /// builder type
        turbo::Status InsertMemoValues(const Array &values) {
            return memo_table_->InsertValues(values);
        }

        /// \brief append a whole dense array to the builder
        template<typename T1 = T>
        enable_if_t<!is_fixed_size_binary_type<T1>::value, turbo::Status> AppendArray(
                const Array &array) {
            using ArrayType = typename TypeTraits<T>::ArrayType;

#ifndef NDEBUG
            TURBO_RETURN_NOT_OK(ArrayBuilder::check_array_type(
                    value_type_, array, "Wrong value type of array to be appended"));
#endif

            const auto &concrete_array = static_cast<const ArrayType &>(array);
            for (int64_t i = 0; i < array.length(); i++) {
                if (array.is_null(i)) {
                    TURBO_RETURN_NOT_OK(append_null());
                } else {
                    TURBO_RETURN_NOT_OK(append(concrete_array.get_view(i)));
                }
            }
            return turbo::OkStatus();
        }

        template<typename T1 = T>
        enable_if_fixed_size_binary<T1, turbo::Status> AppendArray(const Array &array) {
#ifndef NDEBUG
            TURBO_RETURN_NOT_OK(ArrayBuilder::check_array_type(
                    value_type_, array, "Wrong value type of array to be appended"));
#endif

            const auto &concrete_array = static_cast<const FixedSizeBinaryArray &>(array);
            for (int64_t i = 0; i < array.length(); i++) {
                if (array.is_null(i)) {
                    TURBO_RETURN_NOT_OK(append_null());
                } else {
                    TURBO_RETURN_NOT_OK(append(concrete_array.get_value(i)));
                }
            }
            return turbo::OkStatus();
        }

        void reset() override {
            // Perform a partial reset. Call ResetFull to also reset the accumulated
            // dictionary values
            ArrayBuilder::reset();
            indices_builder_.reset();
        }

        /// \brief reset and also clear accumulated dictionary values in memo table
        void ResetFull() {
            reset();
            memo_table_.reset(new internal::DictionaryMemoTable(pool_, value_type_));
        }

        turbo::Status resize(int64_t capacity) override {
            TURBO_RETURN_NOT_OK(check_capacity(capacity));
            capacity = std::max(capacity, kMinBuilderCapacity);
            TURBO_RETURN_NOT_OK(indices_builder_.resize(capacity));
            capacity_ = indices_builder_.capacity();
            return turbo::OkStatus();
        }

        /// \brief Return dictionary indices and a delta dictionary since the last
        /// time that finish or FinishDelta were called, and reset state of builder
        /// (except the memo table)
        turbo::Status FinishDelta(std::shared_ptr<Array> *out_indices,
                                  std::shared_ptr<Array> *out_delta) {
            std::shared_ptr<ArrayData> indices_data;
            std::shared_ptr<ArrayData> delta_data;
            TURBO_RETURN_NOT_OK(FinishWithDictOffset(delta_offset_, &indices_data, &delta_data));
            *out_indices = make_array(indices_data);
            *out_delta = make_array(delta_data);
            return turbo::OkStatus();
        }

        /// \cond FALSE
        using ArrayBuilder::finish;
        /// \endcond

        turbo::Status finish(std::shared_ptr<DictionaryArray> *out) { return finish_typed(out); }

        std::shared_ptr<DataType> type() const override {
            return ::nebula::dictionary(indices_builder_.type(), value_type_);
        }

    protected:
        template<typename c_type>
        turbo::Status AppendArraySliceImpl(const typename TypeTraits<T>::ArrayType &dict,
                                           const ArraySpan &array, int64_t offset, int64_t length) {
            const c_type *values = array.get_values<c_type>(1) + offset;
            return VisitBitBlocks(
                    array.buffers[0].data, array.offset + offset, length,
                    [&](const int64_t position) {
                        const int64_t index = static_cast<int64_t>(values[position]);
                        if (dict.is_valid(index)) {
                            return append(dict.get_view(index));
                        }
                        return append_null();
                    },
                    [&]() { return append_null(); });
        }

        template<typename IndexType>
        turbo::Status AppendScalarImpl(const typename TypeTraits<T>::ArrayType &dict,
                                       const Scalar &index_scalar, int64_t n_repeats) {
            using ScalarType = typename TypeTraits<IndexType>::ScalarType;
            const auto index = turbo::checked_cast<const ScalarType &>(index_scalar).value;
            if (index_scalar.is_valid && dict.is_valid(index)) {
                const auto &value = dict.get_view(index);
                for (int64_t i = 0; i < n_repeats; i++) {
                    TURBO_RETURN_NOT_OK(append(value));
                }
                return turbo::OkStatus();
            }
            return append_nulls(n_repeats);
        }

        turbo::Status finish_internal(std::shared_ptr<ArrayData> *out) override {
            std::shared_ptr<ArrayData> dictionary;
            TURBO_RETURN_NOT_OK(FinishWithDictOffset(/*offset=*/0, out, &dictionary));

            // Set type of array data to the right dictionary type
            (*out)->type = type();
            (*out)->dictionary = dictionary;
            return turbo::OkStatus();
        }

        turbo::Status FinishWithDictOffset(int64_t dict_offset,
                                           std::shared_ptr<ArrayData> *out_indices,
                                           std::shared_ptr<ArrayData> *out_dictionary) {
            // Finalize indices array
            TURBO_RETURN_NOT_OK(indices_builder_.finish_internal(out_indices));

            // Generate dictionary array from hash table contents
            TURBO_RETURN_NOT_OK(memo_table_->GetArrayData(dict_offset, out_dictionary));
            delta_offset_ = memo_table_->size();

            // Update internals for further uses of this DictionaryBuilder
            ArrayBuilder::reset();
            return turbo::OkStatus();
        }

        std::unique_ptr<DictionaryMemoTable> memo_table_;

        // The size of the dictionary memo at last invocation of finish, to use in
        // FinishDelta for computing dictionary deltas
        int32_t delta_offset_;

        // Only used for FixedSizeBinaryType
        int32_t byte_width_;

        BuilderType indices_builder_;
        std::shared_ptr<DataType> value_type_;
    };

    template<typename BuilderType>
    class DictionaryBuilderBase<BuilderType, NullType> : public ArrayBuilder {
    public:
        template<typename B = BuilderType>
        DictionaryBuilderBase(
                enable_if_t<std::is_base_of<AdaptiveIntBuilderBase, B>::value, uint8_t>
                start_int_size,
                const std::shared_ptr<DataType> &value_type,
                MemoryPool *pool = default_memory_pool())
                : ArrayBuilder(pool), indices_builder_(start_int_size, pool) {}

        explicit DictionaryBuilderBase(const std::shared_ptr<DataType> &value_type,
                                       MemoryPool *pool = default_memory_pool())
                : ArrayBuilder(pool), indices_builder_(pool) {}

        explicit DictionaryBuilderBase(const std::shared_ptr<DataType> &index_type,
                                       const std::shared_ptr<DataType> &value_type,
                                       MemoryPool *pool = default_memory_pool())
                : ArrayBuilder(pool), indices_builder_(index_type, pool) {}

        template<typename B = BuilderType>
        explicit DictionaryBuilderBase(
                enable_if_t<std::is_base_of<AdaptiveIntBuilderBase, B>::value, uint8_t>
                start_int_size,
                MemoryPool *pool = default_memory_pool())
                : ArrayBuilder(pool), indices_builder_(start_int_size, pool) {}

        explicit DictionaryBuilderBase(MemoryPool *pool = default_memory_pool())
                : ArrayBuilder(pool), indices_builder_(pool) {}

        explicit DictionaryBuilderBase(const std::shared_ptr<Array> &dictionary,
                                       MemoryPool *pool = default_memory_pool())
                : ArrayBuilder(pool), indices_builder_(pool) {}

        /// \brief append a scalar null value
        turbo::Status append_null() final {
            length_ += 1;
            null_count_ += 1;

            return indices_builder_.append_null();
        }

        turbo::Status append_nulls(int64_t length) final {
            length_ += length;
            null_count_ += length;

            return indices_builder_.append_nulls(length);
        }

        turbo::Status append_empty_value() final {
            length_ += 1;

            return indices_builder_.append_empty_value();
        }

        turbo::Status append_empty_values(int64_t length) final {
            length_ += length;

            return indices_builder_.append_empty_values(length);
        }

        /// \brief append a whole dense array to the builder
        turbo::Status AppendArray(const Array &array) {
#ifndef NDEBUG
            TURBO_RETURN_NOT_OK(ArrayBuilder::check_array_type(
                    Type::NA, array, "Wrong value type of array to be appended"));
#endif
            for (int64_t i = 0; i < array.length(); i++) {
                TURBO_RETURN_NOT_OK(append_null());
            }
            return turbo::OkStatus();
        }

        turbo::Status resize(int64_t capacity) override {
            TURBO_RETURN_NOT_OK(check_capacity(capacity));
            capacity = std::max(capacity, kMinBuilderCapacity);

            TURBO_RETURN_NOT_OK(indices_builder_.resize(capacity));
            capacity_ = indices_builder_.capacity();
            return turbo::OkStatus();
        }

        turbo::Status finish_internal(std::shared_ptr<ArrayData> *out) override {
            TURBO_RETURN_NOT_OK(indices_builder_.finish_internal(out));
            (*out)->type = dictionary((*out)->type, null());
            (*out)->dictionary = NullArray(0).data();
            return turbo::OkStatus();
        }

        /// \cond FALSE
        using ArrayBuilder::finish;
        /// \endcond

        turbo::Status finish(std::shared_ptr<DictionaryArray> *out) { return finish_typed(out); }

        std::shared_ptr<DataType> type() const override {
            return ::nebula::dictionary(indices_builder_.type(), null());
        }

    protected:
        BuilderType indices_builder_;
    };

}  // namespace nebula::internal

namespace nebula {
    /// \brief A DictionaryArray builder that uses AdaptiveIntBuilder to return the
    /// smallest index size that can accommodate the dictionary indices
    template<typename T>
    class DictionaryBuilder : public internal::DictionaryBuilderBase<AdaptiveIntBuilder, T> {
    public:
        using BASE = internal::DictionaryBuilderBase<AdaptiveIntBuilder, T>;
        using BASE::BASE;

        /// \brief append dictionary indices directly without modifying memo
        ///
        /// NOTE: Experimental API
        turbo::Status AppendIndices(const int64_t *values, int64_t length,
                                    const uint8_t *valid_bytes = nullptr) {
            int64_t null_count_before = this->indices_builder_.null_count();
            TURBO_RETURN_NOT_OK(this->indices_builder_.append_values(values, length, valid_bytes));
            this->capacity_ = this->indices_builder_.capacity();
            this->length_ += length;
            this->null_count_ += this->indices_builder_.null_count() - null_count_before;
            return turbo::OkStatus();
        }
    };

    /// \brief A DictionaryArray builder that always returns int32 dictionary
    /// indices so that data cast to dictionary form will have a consistent index
    /// type, e.g. for creating a ChunkedArray
    template<typename T>
    class Dictionary32Builder : public internal::DictionaryBuilderBase<Int32Builder, T> {
    public:
        using BASE = internal::DictionaryBuilderBase<Int32Builder, T>;
        using BASE::BASE;

        /// \brief append dictionary indices directly without modifying memo
        ///
        /// NOTE: Experimental API
        turbo::Status AppendIndices(const int32_t *values, int64_t length,
                                    const uint8_t *valid_bytes = nullptr) {
            int64_t null_count_before = this->indices_builder_.null_count();
            TURBO_RETURN_NOT_OK(this->indices_builder_.append_values(values, length, valid_bytes));
            this->capacity_ = this->indices_builder_.capacity();
            this->length_ += length;
            this->null_count_ += this->indices_builder_.null_count() - null_count_before;
            return turbo::OkStatus();
        }
    };

    // ----------------------------------------------------------------------
    // Binary / Unicode builders
    // (compatibility aliases; those used to be derived classes with additional
    //  append() overloads, but they have been folded into DictionaryBuilderBase)

    using BinaryDictionaryBuilder = DictionaryBuilder<BinaryType>;
    using StringDictionaryBuilder = DictionaryBuilder<StringType>;
    using BinaryDictionary32Builder = Dictionary32Builder<BinaryType>;
    using StringDictionary32Builder = Dictionary32Builder<StringType>;

    /// @}

}  // namespace nebula
